package curator;

import java.util.concurrent.TimeUnit;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CreateBuilder;
import org.apache.curator.framework.recipes.barriers.DistributedBarrier;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.EnsurePath;
import org.apache.curator.utils.ZKPaths;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.zookeeper.CreateMode;

public class GetStarted {

	public static String zookeeperConnectionString = "192.168.3.19:9090";
	
	public static RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
	
	public static CuratorFramework client = CuratorFrameworkFactory.newClient(zookeeperConnectionString, 10000, 5000, retryPolicy);
	
	static
	{
		client.start();
	}
	
	public static void ensurePath(String path)
	{
		EnsurePath ensurePath = new EnsurePath(path);

		try {
			ensurePath.ensure(client.getZookeeperClient());		
//			client.setData().forPath(path, "hello".getBytes());
//			client.create().forPath("/my/path", "hello".getBytes());
//			client.delete().forPath("/my/path");
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	
	public static void close()
	{
		client.close();
	}
	
	
	public static void setData(String path, byte[] data)
	{
		try {
			client.setData().forPath(path, data);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public static void create(String path, byte[] data, CreateMode mode)
	{
		try {
			CreateBuilder builder = client.create();
			builder.withMode(mode);
			builder.forPath(path, data);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public static void distributedLock(String processName)
	{
		String lockPath = "/my/lock";
		
		InterProcessMutex lock = new InterProcessMutex(client, lockPath);
		try {
			System.out.println("try to acquire lock ...");
			
			if ( lock.acquire(3600, TimeUnit.SECONDS) ) 
			{
				System.out.println("lock acquired!");
			    try 
			    {
			        for(int i = 0 ; i < 30; ++ i)
			        {
			        	System.out.println(processName);
			        	Thread.sleep(1000);
			        }
			    }
			    finally
			    {
			        lock.release();
			        System.out.println("lock release!");
			    }
			}
			else 
				System.out.println("failed to acquire lock! time expired!");
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		
		client.close();
	}
	
	
	public static void leaderElection()
	{
		String path = "/my/election";
		
		LeaderSelectorListener listener = new LeaderSelectorListenerAdapter()
		{
		    public void takeLeadership(CuratorFramework client) throws Exception
		    {
		        System.out.println("I am the leader now!");
		        Thread.sleep(5000);
		    }
		};

		LeaderSelector selector = new LeaderSelector(client, path, listener);
		selector.autoRequeue();  // not required, but this is behavior that you will probably expect
		selector.start();
		
		
		try {
			Thread.sleep(1000000);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	
	public static void barrierWait()
	{
		String barrierPath = "/my/barrier";
		
		DistributedBarrier barrier = new DistributedBarrier(client, barrierPath);
		
		try {
			System.out.println("waitOnBarrier ... ");
			barrier.waitOnBarrier();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		System.out.println("pass the barrier!");
	}
	
	public static void barrierSet()
	{
		String barrierPath = "/my/barrier";
		
		DistributedBarrier barrier = new DistributedBarrier(client, barrierPath);
		
		try {
			barrier.setBarrier();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		System.out.println("set the barrier!");
	}
	
	public static void barrierRemove()
	{
		String barrierPath = "/my/barrier";
		
		DistributedBarrier barrier = new DistributedBarrier(client, barrierPath);
		
		try {
			barrier.removeBarrier();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		System.out.println("set the barrier!");
	}
	
	public static void startNodeCache(String path)
	{
		final NodeCache cache = new NodeCache(client, path);
		
		try {
			cache.start();
			
			// a PathChildrenCacheListener is optional. Here, it's used just to log changes
	        NodeCacheListener listener = new NodeCacheListener()
	        {
				@Override
				public void nodeChanged() throws Exception {
					System.out.println("node changed!");
					ChildData d = cache.getCurrentData();
		        	
		        	if(d != null)
		        	{
			        	String v = d.getData() == null ? "" : new String(d.getData());
			        	
			        	System.out.println(cache.getCurrentData().getPath() + ":" + v);
		        	}
				}
	        };
	        cache.getListenable().addListener(listener);
	        while(true)
	        {
	        	ChildData d = cache.getCurrentData();
	        	
	        	if(d != null)
	        	{
		        	String v = d.getData() == null ? "" : new String(d.getData());
		        	
		        	System.out.println(cache.getCurrentData().getPath() + ":" + v);
	        	}
	        	Thread.sleep(1000);
	        }
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public static void startPathChildrenCache(String path)
	{
		
        try {
        	
        	PathChildrenCache cache = new PathChildrenCache(client, path, true);
        	
			cache.start();
			
			// a PathChildrenCacheListener is optional. Here, it's used just to log changes
	        PathChildrenCacheListener listener = new PathChildrenCacheListener()
	        {
	            @Override
	            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception
	            {
	                switch ( event.getType() )
	                {
	                    case CHILD_ADDED:
	                    {
	                        System.out.println("Node added: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
	                        break;
	                    }

	                    case CHILD_UPDATED:
	                    {
	                        System.out.println("Node changed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
	                        break;
	                    }

	                    case CHILD_REMOVED:
	                    {
	                        System.out.println("Node removed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
	                        break;
	                    }
	                    default:
	                }
	            }
	        };
	        cache.getListenable().addListener(listener);
	        while(true)
	        {
	        	System.out.print("[");
	        	for(ChildData d : cache.getCurrentData())
	        	{
	        		System.out.print(d.getPath() + ":" + new String(d.getData()) + " ");
	        	}
	        	System.out.println("]");
	        	Thread.sleep(1000);
	        }
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
//	public static void 
	
	public static void main(String[] args) throws Exception {
//		String config = "policy=priority\ninstances=local_jay;1;2\n";
//		String path = "/service/com.to8to.thrift.crm.biddb.BidDbService";
//		setData("/config/esb/flush_stat", (System.currentTimeMillis() + "").getBytes());
		byte[] d = client.getData().forPath("/config/esb/get_stat");
		System.out.println(new String(d));
	}

}
